package com.steakliu.sun.server.netty.handler;

import com.alibaba.fastjson.JSONObject;
import com.steakliu.sun.buffer.api.DataBuffer;
import com.steakliu.sun.buffer.api.DataBufferChannelManager;
import com.steakliu.sun.cache.result.CacheHelper;
import com.steakliu.sun.cache.result.CacheInfo;
import com.steakliu.sun.cache.result.ResultCacheManager;
import com.steakliu.sun.common.helper.SunServiceHelper;
import com.steakliu.sun.common.thread.DataBufferThreadPool;
import com.steakliu.sun.common.thread.ServerConcurrentThreadPool;
import com.steakliu.sun.connections.ConnectionsManager;
import com.steakliu.sun.constant.RateLimiterFailStrategyConstant;
import com.steakliu.sun.constant.SunConstant;
import com.steakliu.sun.invoker.api.Invoker;
import com.steakliu.sun.protocol.*;
import com.steakliu.sun.ratelimiter.api.SunRateLimiter;
import com.steakliu.sun.server.api.ServerConfig;
import com.steakliu.sun.server.api.ServerLocalServiceCaches;
import com.steakliu.sun.server.netty.manager.ServerChannelManager;
import com.steakliu.sun.spi.loader.ExtensionLoader;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

/**
 * 功能说明： server handler
 * <p>
 * Original @Author: steakliu-刘牌, 2023-01-08  12:37
 * <p>
 * Copyright (C)2020-2022  steakliu All rights reserved.
 */
public class SunServerHandler extends SimpleChannelInboundHandler<SunProtocol<SunRequest>> {

    private final Logger LOGGER = LoggerFactory.getLogger(SunServerHandler.class);

    private final Map<String, ServerLocalServiceCaches> localCacheServices;

    private final Invoker invoker;

    private final ServerConfig serverConfig;

    private final ServerConcurrentThreadPool serverConcurrentThreadPool;

    private final ConnectionsManager connectionsManager;

    private SunRateLimiter rateLimiter;

    private DataBuffer dataBuffer;

    private DataBufferChannelManager dataBufferChannelManager;

    private ResultCacheManager resultCacheManager;

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);
        connectionsManager.add(ctx.channel());
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        super.channelUnregistered(ctx);
        connectionsManager.remove(ctx.channel());
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        connectionsManager.add(ctx.channel());
        super.channelActive(ctx);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        connectionsManager.remove(ctx.channel());
    }

    public SunServerHandler(Map<String, ServerLocalServiceCaches> localCacheServices, Invoker invoker, ServerConfig config, ServerConcurrentThreadPool serverConcurrentThreadPool) {
        this.serverConfig = config;
        this.localCacheServices = localCacheServices;
        this.invoker = invoker;
        this.serverConcurrentThreadPool = serverConcurrentThreadPool;
        this.connectionsManager = ConnectionsManager.getInstance(config.getMaxConnections(), config.getConnectionsDisuseStrategyType());
        if (config.isUseRateLimit()) {
            this.rateLimiter = ExtensionLoader.getExtension(SunRateLimiter.class, config.getRateLimiterType());
            this.rateLimiter.init(config.getMilliseconds(), config.getPermits());
        }
        if (config.isEnableResultCache()) {
            this.resultCacheManager = ResultCacheManager.getInstance();
        }
        if (config.isEnableBuffer()) {
            this.dataBuffer = ExtensionLoader.getExtension(DataBuffer.class, config.getBufferType());
            this.dataBuffer.init(1000);
            this.dataBufferChannelManager = DataBufferChannelManager.getInstance();
            DataBufferThreadPool.submit(() -> {
                this.handleRequestBuffer();
            });
        }
    }

    /**
     * 从缓冲区获取缓冲的请求并进行处理
     */
    private void handleRequestBuffer() {
        while (true) {
            try {
                SunProtocol protocol = dataBuffer.take();
                if (null != protocol) {
                    LOGGER.info("消费缓冲区来自【客户端】的请求：{}", JSONObject.toJSONString(protocol));
                    try {
                        ChannelHandlerContext context = dataBufferChannelManager.remove(protocol.getMsgHeader().getRequestId());
                        if (null != context) {
                            this.handleRequestRateLimiterOrCache(protocol, context);
                        }
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                }
                //TODO 稍微睡眠，减轻CPU压力
                Thread.sleep(1);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 事件触发
     *
     * @param ctx
     * @param event
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object event) throws Exception {
        //心跳事件，如果在规定时间没有收到对应的事件，就会触发userEventTriggered
        if (event instanceof IdleStateEvent) {
            try {
                LOGGER.info("Timeout to obtain heartbeat information of client, close channel : {}", ctx.channel().remoteAddress());
                connectionsManager.remove(ctx.channel());
                ctx.channel().close();
            } finally {
                ctx.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
            }
        }
        super.userEventTriggered(ctx, event);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext context, SunProtocol<SunRequest> protocol) {
        serverConcurrentThreadPool.submit(() -> {
            connectionsManager.update(context.channel());
            this.handle(protocol, context);
        });
    }

    /**
     * 处理请求
     */
    private void handle(SunProtocol<SunRequest> protocol, ChannelHandlerContext context) {
        MsgHeader header = protocol.getMsgHeader();
        MsgType msgType = MsgType.findType(header.getType());
        if (msgType == null) {
            return;
        }
        switch (msgType) {
            case REQUEST:
                this.handleRequestRateLimiterOrBuffer(protocol, context);
                break;
            case CLIENT_HEARTBEAT:
                this.handleFromClientHeartBeat(protocol, context);
                break;
            default:
                throw new IllegalStateException("Unexpected value: " + msgType);
        }
    }

    /**
     * 响应来自客户端的心跳
     *
     * @param protocol
     * @return
     */
    private void handleFromClientHeartBeat(SunProtocol<SunRequest> protocol, ChannelHandlerContext context) {
        SunResponse response = new SunResponse();
        MsgHeader msgHeader = protocol.getMsgHeader();
        msgHeader.setType((byte) MsgType.SERVER_RESPONSE_HEARTBEAT.getType());
        response.setResult(SunConstant.PONG);
        SunProtocol<SunResponse> responseSunProtocol = new SunProtocol<>();
        responseSunProtocol.setMsgHeader(msgHeader);
        responseSunProtocol.setBody(response);
        context.writeAndFlush(responseSunProtocol);
    }

    private void handleRequestRateLimiterOrBuffer(SunProtocol<SunRequest> protocol, ChannelHandlerContext context) {
        if (serverConfig.isEnableBuffer()) {
            dataBuffer.push(protocol);
            dataBufferChannelManager.add(protocol.getMsgHeader().getRequestId(), context);
        } else {
            this.handleRequestRateLimiterOrCache(protocol, context);
        }
    }

    private void handleRequestRateLimiterOrCache(SunProtocol<SunRequest> protocol, ChannelHandlerContext context) {
        SunProtocol<SunResponse> responseSunProtocol;
        if (serverConfig.isEnableResultCache()) {
            String cacheKey = CacheHelper.buildCacheKey(protocol.getBody());
            CacheInfo cacheInfo = resultCacheManager.get(cacheKey);
            if (null != cacheInfo) {
                responseSunProtocol = new SunProtocol<>();
                MsgHeader msgHeader = protocol.getMsgHeader();
                msgHeader.setType((byte) MsgType.RESPONSE.getType());
                responseSunProtocol.setMsgHeader(msgHeader);
                SunResponse response = new SunResponse(cacheInfo.getData(), null);
                responseSunProtocol.setBody(response);
            } else {
                responseSunProtocol = this.handleRequest(protocol);
                if (null != responseSunProtocol.getBody().getResult()) {
                    cacheInfo = new CacheInfo(CacheHelper.getExpiredTimestamp(serverConfig.getResultCacheTime()), responseSunProtocol.getBody().getResult());
                    //保存返回结果到缓存
                    resultCacheManager.put(cacheKey, cacheInfo);
                }
            }
        } else {
            responseSunProtocol = this.handleRequestRateLimiter(protocol, context);
        }
        context.writeAndFlush(responseSunProtocol);
    }


    private SunProtocol<SunResponse> handleRequestRateLimiter(SunProtocol<SunRequest> protocol, ChannelHandlerContext context) {
        if (serverConfig.isUseRateLimit()) {
            try {
                if (rateLimiter.tryAcquire()) {
                    return this.handleRequest(protocol);
                } else {
                    if (serverConfig.getRateLimiterFailStrategy().equals(RateLimiterFailStrategyConstant.FALL_BACK)) {
                        //降级处理
                        return this.handleRateLimiterFailFallback(protocol);
                    } else if (serverConfig.getRateLimiterFailStrategy().equals(RateLimiterFailStrategyConstant.RETRY)) {
                        //重试
                        return this.handleRequest(protocol);
                    }
                }
            } finally {
                rateLimiter.release();
            }
        } else {
            return this.handleRequest(protocol);
        }
        return null;
    }


    /**
     * 限流触发降级
     *
     * @param protocol
     * @return
     */
    private SunProtocol<SunResponse> handleRateLimiterFailFallback(SunProtocol<SunRequest> protocol) {
        SunProtocol<SunResponse> responseSunProtocol = new SunProtocol<>();
        MsgHeader header = protocol.getMsgHeader();
        header.setType((byte) MsgType.RESPONSE.getType());
        header.setStatus((byte) MsgStatus.FAIL.getStatus());
        responseSunProtocol.setMsgHeader(header);
        SunResponse response = new SunResponse();
        response.setErrorMsg("server trigger service fallback");
        responseSunProtocol.setBody(response);
        return responseSunProtocol;
    }

    private SunProtocol<SunResponse> handleRequest(SunProtocol<SunRequest> protocol) {
        SunRequest request = protocol.getBody();
        MsgHeader header = protocol.getMsgHeader();
        String serviceKey = SunServiceHelper.getServiceKey(request.getClassName(), request.getVersion(), request.getGroup());
        SunProtocol<SunResponse> sunProtocol = new SunProtocol<>();
        SunResponse response = new SunResponse();
        header.setType((byte) MsgType.RESPONSE.getType());
        try {
            //反射调用
            Object result = this.invoke(serviceKey, request);

            response.setResult(result);
            header.setStatus((byte) MsgStatus.SUCCESS.getStatus());
        } catch (Throwable throwable) {
            throwable.printStackTrace();
            response.setErrorMsg(throwable.getCause().getMessage());
            header.setStatus((byte) MsgStatus.FAIL.getStatus());
            LOGGER.error("Failed to call service : {}", serviceKey);
        }
        sunProtocol.setMsgHeader(header);
        sunProtocol.setBody(response);
        return sunProtocol;
    }

    private Object invoke(String serviceKey, SunRequest request) throws Throwable {
        Object serviceBean = localCacheServices.get(serviceKey).getServiceBean();
        if (null == serviceBean) {
            throw new RuntimeException("no services available , serviceName is " + serviceKey);
        }
        return invoker.invoke(serviceBean, serviceBean.getClass(), request.getMethodName(),
                request.getParameterTypes(), request.getParameters());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        LOGGER.info("server handler exception : ", cause);
        ServerChannelManager.remove(ctx.channel());
        connectionsManager.remove(ctx.channel());
        ctx.close();
    }
}
